#![allow(dead_code)]

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use std::panic::{self, UnwindSafe, RefUnwindSafe};
use std::future::Future;
use std::error::Error;
use log::{info, error};
use tokio::runtime::{Builder, Runtime};
use tokio::sync::{Notify};

// 使用 OnceLock 管理全局实例
static GLOBAL_TASK_MANAGER: OnceLock<Arc<TaskManager>> = OnceLock::new();

/// 任务管理器，用于统一管理多个后台任务
pub struct TaskManager {
    pub running: Arc<AtomicBool>,
    shutdown_notify: Arc<Notify>,
    rt: Arc<Runtime>,
    // 新增专用阻塞线程池
    blocking_rt: Arc<Runtime>,
}

impl TaskManager {
    /// 创建一个新的任务管理器
    pub fn new(rt: &Arc<Runtime>) -> Self {
        // 创建专用的阻塞操作线程池（固定线程数）
        let blocking_rt = Arc::new(
            Builder::new_multi_thread()
                .worker_threads(10) // 固定线程数，避免资源耗尽
                .enable_all()
                .build().unwrap()
        );
        Self {
            running: Arc::new(AtomicBool::new(true)),
            shutdown_notify: Arc::new(Notify::new()),
            rt: Arc::clone(rt),
            blocking_rt
        }
    }

    /// 初始化全局 TaskManager 实例
    pub fn init_global(rt: &Arc<Runtime>) -> Arc<TaskManager> {
        let manager = Arc::new(TaskManager::new(rt));
        GLOBAL_TASK_MANAGER.get_or_init(|| manager.clone());
        manager
    }

    /// 获取全局 TaskManager 实例
    pub fn global() -> Arc<TaskManager> {
        GLOBAL_TASK_MANAGER.get().cloned().expect(
            "Global TaskManager not initialized. Call TaskManager::init_global() first"
        )
    }

    /// 获取共享的运行状态标志
    pub fn get_running_flag(&self) -> Arc<AtomicBool> {
        self.running.clone()
    }

    /// 启动一个周期性任务
    pub fn spawn_periodic<F, Fut>(&self, interval: Duration, task: F)
        where
            F: Fn() -> Fut + Send + Sync + UnwindSafe + RefUnwindSafe + 'static,
            Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,
    {
        let running = self.running.clone();
        let shutdown_notify = self.shutdown_notify.clone();
        let rt = self.rt.clone();

        self.rt.spawn(async move {
            let mut interval_timer = tokio::time::interval(interval);

            loop {
                tokio::select! {
                    // 优先检查关闭通知
                    _ = shutdown_notify.notified() => break,
                    _ = interval_timer.tick() => {
                        // 重构后的任务执行逻辑
                        let result = execute_task(&task, &rt).await;

                        if let Err(e) = result {
                            error!("【周期任务】执行错误: {e}");
                        }

                        // 双重检查运行状态
                        if !running.load(Ordering::SeqCst) {
                            break;
                        }
                    }
                }
            }
            info!("【周期任务】已正常退出");
        });
    }

    pub fn spawn_periodic_after_complete<F, Fut>(&self, delay: Duration, task: F)
        where
            F: Fn() -> Fut + Send + Sync + UnwindSafe + RefUnwindSafe + 'static,
            Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,
    {
        let running = self.running.clone();
        let shutdown_notify = self.shutdown_notify.clone();
        let rt = self.rt.clone();

        self.rt.spawn(async move {
            loop {
                // 执行任务
                let result = execute_task(&task, &rt).await;

                if let Err(e) = result {
                    error!("【周期任务】执行错误: {e}");
                }

                // 双重检查运行状态
                if !running.load(Ordering::SeqCst) {
                    break;
                }

                // 等待通知触发或延迟结束
                tokio::select! {
                _ = shutdown_notify.notified() => {
                    info!("【周期任务】收到关闭通知，退出循环");
                    break;
                }
                _ = tokio::time::sleep(delay) => {
                    // 延迟结束后继续下一次循环
                }
            }
            }

            info!("【周期任务】已正常退出");
        });
    }

    /// 通用任务调度函数（带日志）
    pub fn spawn_once_task<F, Fut>(&self,task_name: &'static str, task_func: F)
        where
            F: FnOnce() -> Fut + Send + std::panic::UnwindSafe + std::panic::RefUnwindSafe + 'static,
            Fut: Future<Output = Result<bool, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
    {
        self.spawn_once(move || async move {
            match task_func().await {
                Ok(false) => {
                    log::warn!("[{}] 跳过执行", task_name);
                }
                Ok(true) => {
                    log::info!("[{}] 执行成功", task_name);
                }
                Err(e) => {
                    log::error!("[{}] 执行失败: {:?}", task_name, e);
                }
            }

            Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
        });
    }

    /// 启动一个一次性任务
    pub fn spawn_once<F, Fut>(&self, task: F)
        where
            F: FnOnce() -> Fut + Send + UnwindSafe + RefUnwindSafe + 'static,
            Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,
    {
        let rt = self.rt.clone();

        self.rt.spawn(async move {
            let result = execute_task_once(task, &rt).await;

            match result {
                Ok(()) => info!("【一次性任务】执行成功"),
                Err(e) => error!("【一次性任务】执行错误: {e}"),
            }
        });
    }

    /// 通知所有任务优雅退出
    pub fn shutdown(&self) {
        self.running.store(false, Ordering::SeqCst);
        self.shutdown_notify.notify_waiters();
        info!("【任务管理器】已发送关闭信号，等待任务退出...");
    }

    /// 内部通用阻塞执行函数
    fn execute_blocking<R, F>(&self, f: F) -> Option<R>
        where
            F: FnOnce() -> R + Send + 'static,
            R: Send + 'static,
    {
        // 使用标准库线程执行阻塞操作
        let (tx, rx) = std::sync::mpsc::channel();
        let f = Arc::new(Mutex::new(Some(f)));

        // 创建标准库线程执行阻塞操作
        std::thread::spawn(move || {
            let mut f_guard = f.lock().unwrap();
            let f = f_guard.take().unwrap();
            let result = f();
            let _ = tx.send(result);
        });

        rx.recv().ok()
    }

    /// 阻塞当前线程并等待任务完成
    pub fn block_on<F, T>(&self, task: F) -> Option<T>
        where
            F: FnOnce() -> tokio::task::JoinHandle<T> + Send + 'static,
            T: Send + 'static,
    {
        let blocking_rt = self.blocking_rt.clone();
        self.execute_blocking(move || {
            let handle = task();
            match blocking_rt.block_on(handle) {
                Ok(result) => Some(result),
                Err(e) => {
                    error!("【阻塞任务】执行失败: {}", e);
                    None
                }
            }
        }).flatten()
    }

    /// 阻塞当前线程并等待任务完成，并返回结果
    pub fn block_on_result<F, T, E>(&self, task: F) -> Result<T, E>
        where
            F: FnOnce() -> tokio::task::JoinHandle<Result<T, E>> + Send + 'static,
            T: Send + 'static,
            E: From<String> + Send + 'static,
    {
        self.block_on(task)
            .unwrap_or_else(|| Err("任务执行或接收结果失败".to_string().into()))
    }

    /// 阻塞当前线程并等待 Future 完成
    pub fn block_on_future<F, T>(&self, future: F) -> Option<T>
        where
            F: Future<Output = T> + Send + 'static,
            T: Send + 'static,
    {
        let blocking_rt = self.blocking_rt.clone();
        self.execute_blocking(move || {
            blocking_rt.block_on(future)
        })
    }

    /// 阻塞当前线程并等待 Future 完成，返回 Result<T, E>
    pub fn block_on_future_result<F, T, E>(&self, future: F) -> Result<T, E>
        where
            F: Future<Output = Result<T, E>> + Send + 'static,
            T: Send + 'static,
            E: From<String> + Send + 'static,
    {
        self.block_on_future(async move {
            future.await
        })
            .unwrap_or_else(|| Err("任务执行或接收结果失败".to_string().into()))
    }
}

/// 重构后的任务执行函数（处理 panic 和错误）
async fn execute_task<F, Fut>(
    task: &F,
    rt: &Arc<Runtime>
) -> Result<(), Box<dyn Error + Send + Sync>>
    where
        F: Fn() -> Fut + UnwindSafe + RefUnwindSafe,
        Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,
{
    // 捕获同步 panic
    let future_result = panic::catch_unwind(panic::AssertUnwindSafe(|| task()));

    match future_result {
        Ok(future) => {
            // 使用 runtime 执行 future 并捕获异步错误
            rt.spawn(future).await?
        }
        Err(panic) => {
            // 处理 panic
            let error_msg = if let Some(s) = panic.downcast_ref::<&str>() {
                s.to_string()
            } else if let Some(s) = panic.downcast_ref::<String>() {
                s.clone()
            } else {
                "未知类型 panic".to_string()
            };
            return Err(Box::new(std::io::Error::new(
                std::io::ErrorKind::Other,
                format!("任务发生 panic: {}", error_msg)
            )));
        }
    }
}

/// 一次性任务的执行函数
async fn execute_task_once<F, Fut>(
    task: F,
    rt: &Arc<Runtime>
) -> Result<(), Box<dyn Error + Send + Sync>>
    where
        F: FnOnce() -> Fut + UnwindSafe + RefUnwindSafe,
        Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,
{
    // 捕获同步 panic
    let future_result = panic::catch_unwind(panic::AssertUnwindSafe(|| task()));

    match future_result {
        Ok(future) => {
            // 使用 runtime 执行 future 并捕获异步错误
            rt.spawn(future).await?
        }
        Err(panic) => {
            // 处理 panic
            let error_msg = if let Some(s) = panic.downcast_ref::<&str>() {
                s.to_string()
            } else if let Some(s) = panic.downcast_ref::<String>() {
                s.clone()
            } else {
                "未知类型 panic".to_string()
            };
            return Err(Box::new(std::io::Error::new(
                std::io::ErrorKind::Other,
                format!("任务发生 panic: {}", error_msg)
            )));
        }
    }
}

/// 统一处理 panic 日志
fn log_panic(panic: &Box<dyn std::any::Any + Send>) {
    let msg = if let Some(s) = panic.downcast_ref::<&str>() {
        s.to_string()
    } else if let Some(s) = panic.downcast_ref::<String>() {
        s.clone()
    } else {
        format!("{:?}", panic)
    };
    error!("【任务】发生 panic: {msg}");
}